
package cn.com.netedge.iot.frontend.comm.mq;

import cn.com.netedge.iot.frontend.comm.constants.AppConstant;
import cn.com.netedge.iot.frontend.comm.constants.ChannelConstant;
import cn.com.netedge.iot.frontend.comm.utils.DeviceLogUtil;
import com.alibaba.fastjson2.JSON;
import cn.com.netedge.iot.common.dataobject.SendFrame;
import cn.com.netedge.iot.common.enums.MessageDirectEnum;
import cn.com.netedge.iot.common.mq.MqMessageListener;
import cn.com.netedge.iot.common.utils.RedisUtil;
import cn.com.netedge.iot.common.utils.StringUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.socket.DatagramPacket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;


@Component
@Scope("prototype")
public class DownMessageListener extends MqMessageListener {
    private static final Logger logger = LoggerFactory.getLogger(DownMessageListener.class);
    private final static int LENGTH_MIN = 12;
    @Autowired
    private DelaySendMessageTask delaySendMessageTask;

    public void execute(String msg) {
        if (StringUtil.isEmpty(msg)) {
            logger.error("messsage is null");
            return;
        }
        logger.debug("message text: {}", msg);
        SendFrame frame = JSON.parseObject(msg, SendFrame.class);
        if (frame == null) {
            logger.error("frame is null");
            return;
        }
        byte[] buf = frame.getData();
        if (buf == null) {
            logger.error("frame's data is null");
            return;
        }
        if (buf.length < LENGTH_MIN) {
            logger.error("frame's data length is not enough");
            return;
        }
        String deviceNumber = frame.getDeviceNumber();
        int deviceType = frame.getDeviceType();
        logger.debug("deviceNumber:{} deviceType:{} send message:{}.", deviceNumber, deviceType, msg);

        try {
            String path = AppConstant.machineName + "/" + deviceNumber + "/" + deviceType;
            String channelId = RedisUtil.get(path);
            if (channelId == null) {
                logger.warn("channelId:{} not exists in redis", path);
                return;
            }
            Channel channel = ChannelConstant.CHANNEL_MAP.get(channelId);
            if (channel == null) {
                logger.error("channelId:{} 's channel not found", channelId);
                return;
            }
            if (!channel.isActive()) {
                logger.error("channelId:{} channel not active", channelId);
                return;
            }
            //保证终端串行执行
            String key = deviceNumber + "_" + deviceType;
            Long lastSendTime = ChannelConstant.SEND_FLAG_MAP.get(key);
            long now = System.currentTimeMillis();
            if (lastSendTime != null && (now - lastSendTime) / 1000 < AppConstant.MESSAGE_SEND_TIMEOUT) {
                delaySendMessageTask.sendMessage(deviceNumber, deviceType, msg);
                logger.error("deviceNumber:{} deviceType:{} has other task", deviceNumber, deviceType);
                return;
            }
            ChannelConstant.SEND_FLAG_MAP.put(key, now);

            if (ChannelConstant.DATAGRAM_PACKET_MAP.containsKey(path)) { //使用udp 发送报文
                DatagramPacket flagPacket = ChannelConstant.DATAGRAM_PACKET_MAP.get(path);
                ByteBuf byteBuf = Unpooled.wrappedBuffer(buf);
                DatagramPacket datagramPacket = new DatagramPacket(byteBuf, flagPacket.sender());
                channel.writeAndFlush(datagramPacket);
            } else {
                channel.writeAndFlush(buf);
            }

            //记录报文日志
            DeviceLogUtil.deviceLogger.logMessage(deviceNumber, deviceType, MessageDirectEnum.DOWN, buf);
        } catch (Exception e) {
            logger.error("consume message error.", e);
        }
    }

    @Override
    public boolean checkCacheQueue() {
        return true;
    }
}
